package lucene.server.web;


import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import lucene.server.cdc.CDCTask;
import lucene.server.common.datasource.PoolManager;
import lucene.server.common.model.TaskConfig;
import lucene.server.common.utils.TaskUtil;
import lucene.server.core.LuceneService;
import lucene.server.core.model.IndexConfig;
import lucene.server.web.config.LuceneConfig;
import lucene.server.web.dao.SystemDao;
import lucene.server.web.model.CDCConfig;
import lucene.server.web.task.DataChangeTask;
import lucene.server.web.task.SQLTask;
import lucene.server.web.task.TaskPool;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
@Slf4j
public class LuceneServerInit implements ApplicationRunner {
    /**
     * 动态任务
     * */
    private static boolean shutdownHookRegistered = false;


    @SneakyThrows
    public   void restart() {
        LuceneService.shutdownAll();
        LuceneConfig.clearConfigs();
        TaskPool.shutdown();
        run(null);
    }
    @SneakyThrows
    public   void restartOne(String indexName) {
        shutdownOne(indexName);
        runOne(indexName);
    }
    @SneakyThrows
    public void shutdownOne(String indexName) {
        LuceneConfig.removeConfig(indexName);
        LuceneService.shutdown(indexName);
        TaskPool.shutdown("change_"+indexName);
        TaskPool.shutdown("sql_"+indexName+"重建任务");
    }
    @SneakyThrows
    public void runOne(String indexName) {
        LuceneConfig.reCreateConfig(indexName);
        IndexConfig indexConfig = LuceneConfig.getIndexConfig(indexName);
        if(indexConfig == null){ return; }
        LuceneService.createIndex(indexConfig);
        TaskPool.addTask(new SQLTask(indexConfig));
        if (indexConfig.isCdc()){
            TaskPool.addTask(new DataChangeTask(indexConfig));
        }
    }
    @Override
    public void run(ApplicationArguments args) throws Exception {

        //初始化配置
        LuceneConfig.createConfigs();

        List<IndexConfig> indexConfigs = LuceneConfig.getIndexConfigs();
        //创建索引
        LuceneService.createIndex(indexConfigs);
        log.info("{} Lucene Index Create Complete！", indexConfigs.size());

        //数据库连接初始化
        PoolManager.poolInit();

        //创建CDC接收任务和索引的SQL任务
        for (IndexConfig indexConfig : indexConfigs){
            if (indexConfig.isCdc()){
                DataChangeTask luceneTask = new DataChangeTask(indexConfig);
                TaskPool.addTask(luceneTask);
            }
        }
        //创建索引的重建任务
        for (IndexConfig indexConfig : indexConfigs){
            SQLTask sqlTask = new SQLTask(indexConfig);
            TaskPool.addTask(sqlTask);
        }

        //创建独立的SQL任务
        List<TaskConfig> taskConfigs = (List<TaskConfig>) SystemDao.getAll("task",TaskConfig.class);
        for (TaskConfig taskConfig : taskConfigs){
            SQLTask sqlTask = new SQLTask(taskConfig);
            TaskPool.addTask(sqlTask);
        }

        //cdc
        List<CDCConfig> cdcConfigs = (List<CDCConfig>) SystemDao.getAll("cdc",CDCConfig.class);
        for (CDCConfig cdcConfig : cdcConfigs){
            CDCTask canalTask = new CDCTask(cdcConfig.toProperties());
            TaskPool.addTask(canalTask);
        }

        //创建flink任务

        //注册ShutdownHook
        if (!shutdownHookRegistered) {
            Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
                @Override
                public void run() {
                    close();
                }
            }));
            shutdownHookRegistered = true;
        }
    }

    /**
     * 系统关闭钩子，主要是为了保证数据的一致性
     * 先停止canal或者flink的数据
     * 然后处理缓存中待更新数据
     * 最后索引commit
     * */
    private void close() {
        log.info("Closing Application resources...");
        TaskPool.shutdown();
        LuceneService.shutdownAll();
    }


}
